package com.shujia.spark.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo1WordCount {
  def main(args: Array[String]): Unit = {

    /**
      *
      * 创建spark sql环境
      *
      */

    val spark: SparkSession = SparkSession
      .builder()
      .appName("wc")
      .master("local")
      .getOrCreate()


    /**
      * 1、读取数据
      *
      */

    /**
      * DataFrame: 表结构
      * 在rdd的基础上增加了列名和列类型
      *
      */


    val linesDF: DataFrame = spark
      .read
      .format("csv") //读取数据的格式
      .option("sep", "\t") //指定字段的分割方式，默认是逗号
      .schema("line STRING") //指定字段名和字段类型
      .load("data/words.txt")


    //将DF注册成一致临时视图
    linesDF.createOrReplaceTempView("lines")


    /**
      * 基于创建的表编写sql
      *
      */


    val wordCountDF: DataFrame = spark.sql(
      """
        |select word,count(1) as c from (
        |select explode(split(line,',')) word from lines
        |) as a
        |group by word
        |
      """.stripMargin)


    wordCountDF.show()


    //将df的数据保存到hdfs中
    wordCountDF
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)//如果输出路径已存在自动覆盖
      .save("data/wc")

  }

}
